package org.example.data_stream;

import org.apache.flink.connector.elasticsearch.sink.Elasticsearch7SinkBuilder;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author shenguangyang
 */
public class E13_SinkToElasticsearch {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        List<Event> events = new ArrayList<Event>() {{
            add(new Event("user02", "./cart", 2000L));
            add(new Event("user02", "./prod?id=10`", 3800L));
            add(new Event("user01", "./home1", 1000L));
            add(new Event("user01", "./prod?id=15`", 4000L));
            add(new Event("user02", "./prod?id=2`", 3000L));
            add(new Event("user03", "./prod?id=20`", 4000L));
            add(new Event("user04", "./prod?id=22`", 4000L));
            add(new Event("user05", "./prod?id=24`", 4000L));
        }};
        DataStream<Event> streamSource = env.fromCollection(events);

        // 自定义类实现 RedisMapper 接口
        streamSource.sinkTo(
                new Elasticsearch7SinkBuilder<Event>()
                        // 下面的设置使 sink 在接收每个元素之后立即提交，否则这些元素将被缓存起来
                        .setBulkFlushMaxActions(1)
                        .setHosts(new HttpHost("work01", 59200, "http"))
                        .setEmitter((element, context, indexer) -> indexer.add(createIndexRequest(element)))
                        .build()
        );

        env.execute();
    }

    private static IndexRequest createIndexRequest(Event element) {
        Map<String, Object> json = new HashMap<>();
        json.put(element.getUser(), element.getUrl());

        return Requests.indexRequest()
                .index("clicks")
                .id(element.getId() + "")
                .source(json);
    }
}
